package cn.doitedu.datayiee_service17;

import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;

/**
 * 下游：数据处理者
 */
class MsgProcessor {
    public void processMsg(String msg){
        // 将收到的消息进行合法性检查

        // 将检查合格的消息按照确定规范进行格式化处理

        // 根据消息中的某指定字段去数据库中查找匹配信息

        // 将匹配到的信息添加到“消息”中

        // ……

    }
}

/**
 * 上游：数据生成者
 */
class GenMessage{
    MsgProcessor processor = new MsgProcessor();
    Random random = new Random();
    public void genMas(){
        while(true){
            // 模拟生成数据
            String msg = "msg "+ random.nextInt(100);

            // 处理数据，此处的处理方案，即为“同步调用”——调用MsgProcessor类的processMsg方法
            // 同步调用的最核心特点是：调用了目标方法后，就得等到目标方法处理完毕并返回，自己的逻辑才能继续往下执行
            processor.processMsg(msg);

            // 处理完成后的其他事情
            // ……
        }
    }
}


/**
 * 中游，缓冲池
 */
class MsgBuffer{
    ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<String>(100);

    public void put(String msg) throws Exception {
        queue.put(msg);
    }

    public String take() throws Exception {
        return queue.take();
    }
}



/**
 * 上游：数据生成者
 */
class GenMessage2{
    MsgBuffer buffer;
    public GenMessage2(MsgBuffer buffer){
        this.buffer = buffer;
    }

    public void genMsg() throws Exception {
        Random random = new Random();
        while(true){
            // 模拟生成数据
            String msg = "msg "+ random.nextInt(100);

            // 将生成的数据放入缓冲
            buffer.put(msg);

            // 继续处理其他事情
            // ……
        }
    }

}

/**
 * 下游：数据处理者
 */
class MsgProcessor2 {
    MsgBuffer buffer;
    public MsgProcessor2(MsgBuffer buffer){
        this.buffer = buffer;
    }
    public void processMsg() throws Exception {
        while(true) {
            // 从缓冲池中获取数据
            String msg = buffer.take();

            // 将收到的消息进行合法性检查
            // 将检查合格的消息按照确定规范进行格式化处理
            // 根据消息中的某指定字段去数据库中查找匹配信息
            // 将匹配到的信息添加到“消息”中
            // ……
        }
    }
}


public class CallDemo{
    public static void main(String[] args) throws Exception {
        MsgBuffer msgBuffer = new MsgBuffer();


        // 启动生产者，生成数据，只管自己生成，不用等待后续的处理
        new Thread(new Runnable() {
            @Override
            public void run() {

                GenMessage2 genner = new GenMessage2(msgBuffer);
                try {
                    // 生产数据
                    genner.genMsg();
                } catch (Exception e) {}
            }
        }).start();

        // 启动消费者，处理数据，只管自己处理，不用关心数据的生产
        new Thread(new Runnable() {
            @Override
            public void run() {

                MsgProcessor2 processor = new MsgProcessor2(msgBuffer);
                try {
                    // 处理数据
                    processor.processMsg();

                } catch (Exception e) {}


            }
        }).start();

        // 接下来，这里可以干任何事情

    }

}